package com.baishancloud.log.douyin.demand1

import com.baishancloud.log.douyin.stateTTL
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

/**
 *
 * @author ziqiang.wang 
 * @date 2022/3/1 16:42
 */
class AggAndOut1(parameterTool: ParameterTool) extends KeyedProcessFunction[ResultKey1, Result1, Result1] {

  var valueState: ValueState[Result1] = _

  override def open(parameters: Configuration): Unit = {
    // 开启状态
    val description: ValueStateDescriptor[Result1] = new ValueStateDescriptor[Result1]("中间聚合结果", TypeInformation.of(classOf[Result1]))
    val stateTtlConfig: StateTtlConfig = StateTtlConfig.newBuilder(Time.hours(parameterTool.getLong(stateTTL, 1)))
      .updateTtlOnCreateAndWrite()
      .neverReturnExpired()
      .cleanupFullSnapshot()
      .build()
    description.enableTimeToLive(stateTtlConfig)
    valueState = getRuntimeContext.getState(description)
  }

  override def processElement(value: Result1, ctx: KeyedProcessFunction[ResultKey1, Result1, Result1]#Context, out: Collector[Result1]): Unit = {
    // 取出上次结果，如果结果不存在，则更新状态，并输出；如果结果存在，则进行累加，并更新状态，并输出。
    val middleResult: Result1 = valueState.value()
    if (middleResult == null) {
      valueState.update(value)
      out.collect(value)
    } else {
      val result: Result1 = Result1(
        middleResult.time,
        middleResult.step,
        middleResult.name,
        middleResult.endpoint,
        middleResult.value + value.value,
        middleResult.tags,
        Fields(middleResult.fields.unusual_value + value.fields.unusual_value)
      )
      valueState.update(result)
      out.collect(result)
    }
  }

}
